package rocketmq

import (
	"context"
	"errors"
	"fmt"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
)

var producerIns rocketmq.Producer

// 注册rocketmq生产者
func InitProducer(addr []string, groupName string, retryTime int) error {
	if retryTime <= 0 {
		retryTime = 0
	}
	var err error
	producerIns, err = rocketmq.NewProducer(
		producer.WithNameServer(addr),
		producer.WithRetry(retryTime),
		producer.WithGroupName(groupName),
		producer.WithVIPChannel(false),
	)
	if err != nil {
		return err
	}

	err = producerIns.Start()
	if err != nil {
		return err
	}

	return nil
}

// 生产数据
func Publish(topic string, body []byte) (*primitive.SendResult, error) {
	if producerIns == nil {
		return nil, errors.New("RocketMq Producer not register")
	}

	result, err := producerIns.SendSync(context.Background(), &primitive.Message{
		Topic: topic,
		Body:  body,
	})

	if err != nil {
		return nil, err
	}
	if result.Status != primitive.SendOK {
		return nil, fmt.Errorf("RocketMq Producer Publish error status:%v", result.Status)
	}

	return result, nil
}
